package com.chencong.transform;

import com.chencong.bean.WaterSensor;
import com.chencong.env.FlinkTableEnv;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * @Author chencong
 * @Description 滚动聚合算子
 * @Date 7:59 下午 2021/8/18
 * @Param
 **/
public class RollAgg {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment streamTableEnvironment = FlinkTableEnv.getStreamTableEnvironment();
        streamTableEnvironment.setParallelism(1);

        SingleOutputStreamOperator<WaterSensor> MapData = streamTableEnvironment
                .readTextFile("/Users/chencong/IdeaProjects/BigData_Learning/flink/input/WaterSensor")
                .map(new MapFunction<String, WaterSensor>() {
                    @Override
                    public WaterSensor map(String s) throws Exception {
                        String[] data = s.split(",");
                        return new WaterSensor(data[0],Integer.valueOf(data[1]),Integer.valueOf(data[2]));
                    }
                });

        KeyedStream<WaterSensor, String> waterSensorStringKeyedStream = MapData.keyBy(sensor -> sensor.getId());
        waterSensorStringKeyedStream.print("原始数据");
        // waterSensorStringKeyedStream.sum("vc").print("sum");
        //waterSensorStringKeyedStream.max("vc").print("max");
        //waterSensorStringKeyedStream.min("vc").print("min");

        streamTableEnvironment.execute();
    }
}
